package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TrafficPageViewBean;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.MyClickHouseUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author:bryant
 * @create:2022-05-30
 * @Des: 流量域版本-渠道-地区-访客类别粒度页面浏览各窗口(页面浏览、浏览时长、会话数量、独立访客数量、跳出会话数量)汇总表
 */
public class DwsTrafficVcChArIsNewPageViewWindow {
    public static void main(String[] args) throws Exception {
        // TODO: 2022/5/30 1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO: 2022/5/30 2.检查点相关设置 


        // TODO: 2022/5/30 3.从kafak三个主题读取数据 
        //声明消费者主题和消费者组
        String pvTopic = "dwd_traffic_page_log";
        String uvTopic ="dwd_trffic_unique_visitor_detail";
        String ujdTopic= "dwd_traffic_user_jump_detail";

        String groupId = "dws_traffic_vc_ch-isnew_group";
        //获取消费者对象
        FlinkKafkaConsumer<String> pvKafkaConsumer = MyKafkaUtil.getKafkaConsumer(pvTopic, groupId);
        FlinkKafkaConsumer<String> uvKafkaConsumer = MyKafkaUtil.getKafkaConsumer(uvTopic, groupId);
        FlinkKafkaConsumer<String> ujdKafkaConsumer = MyKafkaUtil.getKafkaConsumer(ujdTopic, groupId);
        //将消费者对象转换为流
        DataStreamSource<String> pvStrDS = env.addSource(pvKafkaConsumer);
        DataStreamSource<String> uvStrDS = env.addSource(uvKafkaConsumer);
        DataStreamSource<String> ujdStrDS = env.addSource(ujdKafkaConsumer);

//        pvStrDS.print("1111111");
//        uvStrDS.print("2222222");
//        ujdStrDS.print("3333333");
        // TODO: 2022/5/30 4.类型转换
        //pagelog转换
        SingleOutputStreamOperator<TrafficPageViewBean> pvStatsDS = pvStrDS.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String value) throws Exception {
                JSONObject jsonObj = JSON.parseObject(value);
                JSONObject commomObj = jsonObj.getJSONObject("commom");
                JSONObject pageObj = jsonObj.getJSONObject("page");
                String lastPageId = pageObj.getString("last_page_id");
                long svCt = 0;
                if (lastPageId == null || lastPageId.length() ==0){
                    svCt = 1L;
                }
                return new TrafficPageViewBean(
                    "",
                        "",
                        commomObj.getString("vc"),
                        commomObj.getString("ch"),
                        commomObj.getString("ar"),
                        commomObj.getString("is_new"),
                        0L,
                        svCt,
                        1L,
                        pageObj.getLong("during_time"),
                        0L,
                        jsonObj.getLong("ts")

                );
            }
        });
        //uv转换
        SingleOutputStreamOperator<TrafficPageViewBean> uvStatsDS = uvStrDS.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String value) throws Exception {
                JSONObject jsonObj = JSON.parseObject(value);
                JSONObject commomObj = jsonObj.getJSONObject("commom");
                return new TrafficPageViewBean(
                        "",
                        "",
                        commomObj.getString("vc"),
                        commomObj.getString("ch"),
                        commomObj.getString("ar"),
                        commomObj.getString("is_new"),
                        1L,
                        0L,
                        0L,
                        0L,
                        0L,
                        jsonObj.getLong("ts")
                );
            }
        });

        //ujd转换
        SingleOutputStreamOperator<TrafficPageViewBean> ujdStatsDS = ujdStrDS.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String value) throws Exception {
                JSONObject jsonObj = JSON.parseObject(value);
                JSONObject commomObj = jsonObj.getJSONObject("commom");
                return new TrafficPageViewBean(
                        "",
                        "",
                        commomObj.getString("vc"),
                        commomObj.getString("ch"),
                        commomObj.getString("ar"),
                        commomObj.getString("is_new"),
                        0L,
                        0L, 0L, 0L, 1L,
                        jsonObj.getLong("ts")
                );
            }
        });

        // TODO: 2022/5/30 5.合并流
        DataStream<TrafficPageViewBean> unionDS = pvStatsDS.union(uvStatsDS, ujdStatsDS);

        // TODO: 2022/5/30 6.添加水位线
        SingleOutputStreamOperator<TrafficPageViewBean> trafficViewWithWatermarkDS  = unionDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
            @Override
            public long extractTimestamp(TrafficPageViewBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));
        // TODO: 2022/5/30 7.分组 
        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> keyedDS = trafficViewWithWatermarkDS.keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {

            @Override
            public Tuple4<String, String, String, String> getKey(TrafficPageViewBean value) throws Exception {
                return Tuple4.of(
                        value.getVc(),
                        value.getCh(),
                        value.getAr(),
                        value.getIsNew()
                );
            }
        });
        // TODO: 2022/5/30 8.开窗 
        WindowedStream<TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow> windowDS =
                keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        // TODO: 2022/5/30 9.聚合
        //reduce与 aggregate区别：，两者都是增量聚合，前者是输入输出类型一样，后者是可以输出其他类型，但实现方法较多
        //apply 与 process: 两者都是一条数据一条数据的处理吗，全量处理

        SingleOutputStreamOperator<TrafficPageViewBean> reduceDS = windowDS.reduce(new ReduceFunction<TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                value1.setUjCt(value1.getUjCt() + value2.getUjCt());
                return value1;
            }
        }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {

            @Override
            public void apply(Tuple4<String, String, String, String> key, TimeWindow window, Iterable<TrafficPageViewBean> input, Collector<TrafficPageViewBean> out) throws Exception {
                //添加窗口时间
                for (TrafficPageViewBean trafficPageViewBean : input) {
                    trafficPageViewBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                    trafficPageViewBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                    out.collect(trafficPageViewBean);
                }
            }
        });

        // TODO: 2022/5/30 将数据写入到clickhouse中
        reduceDS.addSink(MyClickHouseUtil.getJdbcSink("insert into dws_traffic_vc_ch_ar_is_new_page_view_window values(?,?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
